package com.example.fg.kettle;

import lombok.extern.slf4j.Slf4j;
import org.pentaho.di.core.KettleEnvironment;
import org.pentaho.di.core.database.DatabaseMeta;
import org.pentaho.di.core.exception.KettleException;
import org.pentaho.di.core.exception.KettleXMLException;
import org.pentaho.di.core.plugins.PluginRegistry;
import org.pentaho.di.core.plugins.StepPluginType;
import org.pentaho.di.repository.kdr.KettleDatabaseRepository;
import org.pentaho.di.repository.kdr.KettleDatabaseRepositoryMeta;
import org.pentaho.di.trans.Trans;
import org.pentaho.di.trans.TransHopMeta;
import org.pentaho.di.trans.TransMeta;
import org.pentaho.di.trans.step.StepMeta;
import org.pentaho.di.trans.steps.insertupdate.InsertUpdateMeta;
import org.pentaho.di.trans.steps.tableinput.TableInputMeta;
import org.pentaho.di.trans.steps.tableoutput.TableOutputMeta;

import javax.servlet.http.HttpServletRequest;
import java.io.*;

@Slf4j
public class KettleClient {

    public static void main(String[] args) {
        try {
            KettleClient client = new KettleClient();
            client.initKettleEnvironment(null);

            TransMeta meta = client.buildTransMeta("kettle");
            PluginRegistry registry = client.getRegistry();

            StepMeta step1 = client.setTableInputStep(meta, registry,
                    "test-data", "select id,name from stu1", "table input");

            StepMeta step2 = client.setTableOutput(meta, registry,
                    "test-data", "stu2", "table insert");
            client.addTransHop(meta, step1, step2);

            client.executeTrans(meta, "test");
        } catch (KettleException e) {
            e.printStackTrace();
        }
    }

    /**
     * 初始化环境
     */
    public void initKettleEnvironment(HttpServletRequest request) throws KettleException {
        if (KettleEnvironment.isInitialized()) {
            return;
        }

        if (request == null) {
            // 运行环境初始化
            KettleEnvironment.init();
        } else {
            String userDir = System.getProperty("user.dir");
            String kettleHome = request.getSession().getServletContext().getRealPath(File.separator + "WEB-INF");
            // 设置用户路径和系统环境，包括用户路径和主目录
            System.setProperty("user.dir", kettleHome);
            System.setProperty("KETTLE_HOME", kettleHome);
            // 运行环境初始化
            KettleEnvironment.init();
            // 避免造成影响其他程序的运行，还原用户路径
            System.setProperty("user.dir", userDir);
        }
    }

    /**
     * 创建转化元
     */
    public TransMeta buildTransMeta(String metaName, String... transXML) throws KettleXMLException {
        TransMeta transMeta = new TransMeta();
        // 设置转化元的名称
        transMeta.setName(metaName);
        // 添加转换的数据库连接
        transMeta.addDatabase(getDatabaseMeta());
        return transMeta;
    }

    /**
     * 设置表输入步骤
     */
    public StepMeta setTableInputStep(TransMeta transMeta, PluginRegistry registry, String sourceDbName, String sql, String stepName) {
        // 创建表输入
        TableInputMeta tableInputMeta = new TableInputMeta();
        String pluginId = registry.getPluginId(StepPluginType.class, tableInputMeta);
        // 指定数据源数据库配置名
        DatabaseMeta source = transMeta.findDatabase(sourceDbName);
        tableInputMeta.setDatabaseMeta(source);
        tableInputMeta.setSQL(sql);
        // 将表输入添加到转换中
        StepMeta stepMeta = new StepMeta(pluginId, stepName, tableInputMeta);
        // 给步骤添加在spoon工具中的显示位置
        stepMeta.setDraw(true);
        stepMeta.setLocation(100, 100);
        // 将表输入添加到步骤中
        transMeta.addStep(stepMeta);
        return stepMeta;
    }

    /**
     * 设置表输出步骤，用于整表抽取
     */
    public StepMeta setTableOutput(TransMeta transMeta, PluginRegistry registry, String targetDbName, String targetTableName, String stepName) {
        // 创建表输出
        TableOutputMeta tableOutputMeta = new TableOutputMeta();
        String pluginId = registry.getPluginId(StepPluginType.class, tableOutputMeta);
        // 配置表输出的目标数据库配置名

        DatabaseMeta targetDb = transMeta.findDatabase(targetDbName);
        tableOutputMeta.setDatabaseMeta(targetDb);
        tableOutputMeta.setTableName(targetTableName);
        // 将表输出添加到转换中
        StepMeta stepMeta = new StepMeta(pluginId, stepName, tableOutputMeta);
        transMeta.addStep(stepMeta);
        return stepMeta;
    }

    /**
     * 设置表插入与更新步骤，用于表中部分字段更新
     */
    public StepMeta setInsertUpdateMeta(TransMeta transMeta, PluginRegistry registry, String targetDbName, String targetTableName, String[] updatelookup, String[] updateStream, String[] updateStream2, String[] conditions, Boolean[] updateOrNot, String stepName) {
        // 创建插入与更新
        InsertUpdateMeta insertUpdateMeta = new InsertUpdateMeta();
        String pluginId = registry.getPluginId(StepPluginType.class, insertUpdateMeta);
        // 配置目标数据库配置名
        DatabaseMeta database_target = transMeta.findDatabase(targetDbName);
        insertUpdateMeta.setDatabaseMeta(database_target);
        // 设置目标表名
        insertUpdateMeta.setTableName(targetTableName);
        // 设置用来查询的关键字
        insertUpdateMeta.setKeyLookup(updatelookup);
        insertUpdateMeta.setKeyStream(updateStream);
        insertUpdateMeta.setKeyStream2(updateStream2);// 这一步不能省略
        insertUpdateMeta.setKeyCondition(conditions);
        // 设置要更新的字段
        insertUpdateMeta.setUpdateLookup(updatelookup);
        insertUpdateMeta.setUpdateStream(updateStream);
        insertUpdateMeta.setUpdate(updateOrNot);
        // 添加步骤到转换中
        StepMeta stepMeta = new StepMeta(pluginId, stepName, insertUpdateMeta);
        stepMeta.setDraw(true);
        stepMeta.setLocation(250, 100);
        transMeta.addStep(stepMeta);
        return stepMeta;
    }

    /**
     * 用于将表输入步骤与第二步骤绑定 绑定关联步骤
     */
    public void addTransHop(TransMeta transMeta, StepMeta from, StepMeta to) {
        transMeta.addTransHop(new TransHopMeta(from, to));
    }

    /**
     * 执行抽取
     */
    @SuppressWarnings("resource")
    public void executeTrans(TransMeta transMeta, String targetDbName) {
        try {
            KettleDatabaseRepository repository = RepositoryCon();
            transMeta.setRepository(repository);
            Trans trans = new Trans(transMeta);
            trans.execute(new String[]{"start..."});
            trans.waitUntilFinished();
            // 关闭数据库连接
            if (trans.getErrors() > 0) {
                throw new RuntimeException("There were errors during transformation execution.");
            }
        } catch (KettleException e) {
            e.printStackTrace();
        }
    }

    private static DatabaseMeta getDatabaseMeta() {
        DatabaseMeta databaseMeta = new DatabaseMeta("test-data", "MYSQL", "Native", "localhost",
                "test", "3306", "root", "toor@1234");
        // 关闭mysql推荐SSL连接提示
        databaseMeta.addExtraOption("MYSQL", "useSSL", "false");
        return databaseMeta;
    }

    /**
     * * 连接到资源库
     */
    private static KettleDatabaseRepository RepositoryCon() throws KettleException {

        // 初始化环境
        if (!KettleEnvironment.isInitialized()) {
            try {
                KettleEnvironment.init();
            } catch (KettleException e) {
                e.printStackTrace();
            }
        }
        // 数据库形式的资源库元对象
        KettleDatabaseRepositoryMeta kettleDatabaseRepositoryMeta = new KettleDatabaseRepositoryMeta();
        kettleDatabaseRepositoryMeta.setConnection(getDatabaseMeta());
        // 数据库形式的资源库对象
        KettleDatabaseRepository kettleDatabaseRepository = new KettleDatabaseRepository();
        // 用资源库元对象初始化资源库对象
        kettleDatabaseRepository.init(kettleDatabaseRepositoryMeta);
        // 连接到资源库 ， 默认的连接资源库的用户名和密码
        kettleDatabaseRepository.connect("admin", "admin");
        if (kettleDatabaseRepository.isConnected()) {
            System.out.println("连接成功");
            return kettleDatabaseRepository;
        } else {
            System.out.println("连接失败");
            return null;
        }
    }


    public PluginRegistry getRegistry() {
        // 插件注册，用于注册转换中需要用到的插件
        return PluginRegistry.getInstance();
    }


}
